package com.demo.spark.stream;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.storage.StorageLevel;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.receiver.Receiver;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.Random;

public class StreamCount {
    public static void main(String[] args) throws InterruptedException {

        SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("data frame ");
        JavaSparkContext sparkContext = new JavaSparkContext(sparkConf);
        sparkContext.setLogLevel("ERROR");


        JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkContext, Durations.seconds(1));
        JavaReceiverInputDStream<String> lines  = javaStreamingContext.receiverStream(new MyReceiver());
        lines.count().print();
        JavaDStream<String> stringJavaDStream = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                String[] s1 = s.split(",");
                return Arrays.asList(s1).iterator();
            }
        });

        stringJavaDStream.count().print();


        javaStreamingContext.start();
        javaStreamingContext.awaitTermination();

    }

    static class MyReceiver extends Receiver<String> {
        public MyReceiver(StorageLevel storageLevel) {
            super(storageLevel);
        }
        public MyReceiver(){
            super(StorageLevel.MEMORY_AND_DISK());
        }

        @Override
        public void onStart() {
            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    Random random = new Random(System.currentTimeMillis());
                    while (!isStopped()){
//                        store("a b c d e\n");
                        char[] chars = new char[9];
                        for (int i = 0,index=0; i < 5; i++,index++) {
                            chars[index] = (char)(random.nextInt(5)+65);
                            try {
                                chars[++index] = ',';
                            }catch (ArrayIndexOutOfBoundsException e){

                            }
                        }
                        store(String.valueOf(chars));
                        try {
                            Thread.currentThread().sleep(1);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
            thread.setDaemon(true);
            thread.start();
            System.out.println("receiver started");
        }

        @Override
        public void onStop() {
            System.out.println("receiver stopped");
        }
    }
}
